/*
 * Copyright 2011 J. Auer, J. Andert, F.-J. Jendrusch, U. Celebi
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package de.fuberlin.replication.library;

import java.io.IOException;
import java.util.Vector;

import de.fuberlin.replication.application.Synchronizable;
import de.fuberlin.replication.communication.Communication;
import de.fuberlin.replication.logging.Logger;
import de.fuberlin.replication.storage.MetadataDAO;
import de.fuberlin.replication.storage.Storage;
import de.fuberlin.replication.storage.SubsetDAO;

/**
 * Implementation of the Vector Sets replica synchronization mechanism built in Microsoft’s WinFS replica management
 * system.
 */
public class VectorSets {
	private static final String TAG = "VectorSets";

	private Synchronizable application;
	private Communication communication;
	private Logger log;
	private MetadataDAO metadataDAO;
	private SubsetDAO subsetDAO;

	/**
	 * Constructor for VectorSets Class.
	 * 
	 * @param application Class implementing {@link Synchronizable}
	 */
	public VectorSets(Synchronizable application) {
		this.application = application;
		communication = application.getCommunication();
		log = application.getLogger();
		Storage storage = application.getStorage();
		metadataDAO = storage.getMetadataDAO();
		subsetDAO = storage.getSubsetDAO();
		log.d(TAG, "sync: Class instantiated");
	}

	/**
	 * Report an update to the synchronization framework.
	 * 
	 * @param application Object implementing {@link Synchronizable} for accessing the data
	 * @param object String that identifies the data that has been modified
	 */
	public static void update(Synchronizable application, String object) {
		Logger log = application.getLogger();
		Storage storage = application.getStorage();
		SubsetDAO subsetDAO = storage.getSubsetDAO();
		MetadataDAO metadataDAO = storage.getMetadataDAO();
		String replica = application.getAddress();
		long nextVersion = storage.getVersion();
		Subset set;
		Metadata metadata = metadataDAO.get(object);
		log.d(TAG, "update: start updating object <" + object + "> to local version <" + nextVersion + ">");

		// Create a new set if none exists
		if (!subsetDAO.getAll().iterator().hasNext()) {
			set = subsetDAO.create();
			subsetDAO.save(set);
		}

		/*
		 * If the object is unknown, add it to a random existing Subset. Set the version of the associated metadata to
		 * nextVersion
		 */
		if (metadata == null) {
			set = subsetDAO.getAll().iterator().next();
			set.setObjectNames(set.findObjectNames(metadataDAO));
			metadata = new Metadata(object, new Version(replica, nextVersion));
			log.d(TAG, "update: object added to Subset <" + set.toString() + ">");
		} else {
			set = containsObject(subsetDAO, metadataDAO, object);

			if (set == null)
				set = subsetDAO.getAll().iterator().next();

			set.setObjectNames(set.findObjectNames(metadataDAO));
			metadata.setVersion(new Version(replica, nextVersion));
			log.d(TAG, "update: object found in Subset <" + set.toString() + ">");
		}
		log.d(TAG, "update: old knowledge: " + set.getKnowledge().toString());

		// Merge the new Version into all sets knowledge vector
		for (Subset subset : subsetDAO.getAll()) {
			boolean replicaFound = false;

			for (Version v : subset.getKnowledge())
				if (v.getReplica().equals(replica)) {
					v.setVersion(nextVersion);
					replicaFound = true;
					break;
				}

			// Add an entry for this replica if formerly unknown
			if (!replicaFound)
				subset.getKnowledge().add(new Version(replica, nextVersion));

			log.d(TAG, "update: new knowledge: " + subset.getKnowledge().toString());
			subsetDAO.save(subset);
		}

		metadataDAO.save(set, metadata);
		log.d(TAG, "update: data succesfully saved to storage");
	}

	/**
	 * Synchronize with the remote device.
	 * 
	 * @param application Class implementing {@link Synchronizable}
	 * @return true if synchronization was completed, false otherwise
	 * @throws IOException If some crucial communication error occurred
	 * @throws ClassNotFoundException If some crucial communication error occurred, reading from the underlying
	 *             ObjectStream
	 */
	public static boolean synchronize(Synchronizable application) throws IOException, ClassNotFoundException {
		return new VectorSets(application).synchronize();
	}

	/**
	 * Synchronize with the remote device.
	 * 
	 * @return true if synchronization was completed, false otherwise
	 * @throws IOException If some crucial communication error occurred
	 * @throws ClassNotFoundException If some crucial communication error occurred, reading from the underlying
	 *             ObjectStream
	 */
	protected boolean synchronize() throws IOException, ClassNotFoundException {
		// Decide, who requests/serves first by comparing hardware addresses
		communication.write(application.getAddress());
		String remoteAddress = (String) communication.read();
		boolean isRequestor;
		int cmp = application.getAddress().compareTo(remoteAddress);

		if (cmp > 0)
			isRequestor = false;
		else if (cmp < 0)
			isRequestor = true;
		else
			throw new IOException("Unable to determine role because of equal hardware addresses");

		boolean syncCompleted = true;

		if (!isRequestor) {
			if (!serve())
				syncCompleted = false;
			if (!request())
				syncCompleted = false;
		} else {
			if (!request())
				syncCompleted = false;
			if (!serve())
				syncCompleted = false;
		}

		return syncCompleted;
	}

	private static Subset containsObject(SubsetDAO subsetDAO, MetadataDAO metadataDAO, String objectName) {
		for (Subset subset : subsetDAO.getAll()) {
			subset.setObjectNames(subset.findObjectNames(metadataDAO));

			if (subset.containsObject(objectName))
				return subset;
		}
		return null;
	}

	private Subset containsObject(Vector<Subset[]> set, String object) {
		for (Subset[] subsets : set)
			for (String o : subsets[0].getObjectNames())
				if (object.equals(o))
					return subsets[1];

		return null;
	}

	private static boolean containsVersion(Vector<Subset> set, Metadata object) {
		for (Subset subset : set)
			if (subset.containsObject(object.getName()) && subset.containsVersion(object.getVersion()))
				return true;

		return false;
	}

	private boolean request() throws IOException {
		log.d(TAG, "sync: sending knowledge...");
		sendKnowledge();

		try {
			log.d(TAG, "sync: receiving updates...");
			receiveUpdates();
		} catch (ClassNotFoundException e1) {
			log.e(TAG, e1.getMessage());
			return false;
		} catch (IOException e2) {
			log.e(TAG, e2.getMessage());
			return false;
		} finally {
			log.d(TAG, "sync: merging subsets...");
			union();
		}

		return true;
	}

	private boolean serve() throws IOException, ClassNotFoundException {
		log.d(TAG, "sync: receiving knowledge...");
		Vector<Subset> remoteSet = receiveKnowledge();

		try {
			log.d(TAG, "sync: sending updates...");
			sendRelevantSets(remoteSet);
		} catch (IOException e) {
			e.printStackTrace();
			return false;
		}

		return true;
	}

	private Vector<Subset> receiveKnowledge() throws IOException, ClassNotFoundException {
		Vector<Subset> remoteSet = new Vector<Subset>();
		Subset subset = new Subset((VersionVector) communication.read());
		log.d(TAG, "sync: received subset knowledge: " + subset.getKnowledge().toString());

		while (!subset.getKnowledge().isEmpty()) {
			String object = (String) communication.read();
			log.d(TAG, "sync: received object name <" + object + ">");

			while (!object.equals("")) {
				subset.addObjectName(object);
				object = (String) communication.read();
				log.d(TAG, "sync: received object name <" + object + ">");
			}
			remoteSet.add(subset);
			subset = new Subset((VersionVector) communication.read());
			log.d(TAG, "sync: received subset knowledge: " + subset.getKnowledge().toString());
		}
		return remoteSet;
	}

	private void sendKnowledge() throws IOException {
		for (Subset subset : subsetDAO.getAll()) {
			log.d(TAG, "sync: sending knowledge: " + subset.getKnowledge().toString());
			communication.write(subset.getKnowledge());

			for (Metadata object : metadataDAO.getAll(subset)) {
				log.d(TAG, "sync: sending object name <" + object.getName() + ">");
				communication.write(object.getName());
			}

			String s = "";
			log.d(TAG, "sync: sending object name <" + s + ">");
			communication.write(s);
		}
		VersionVector v = new VersionVector();
		log.d(TAG, "sync: sending knowledge <" + v.toString() + ">");
		communication.write(v);
	}

	private void sendRelevantSets(Vector<Subset> remoteSet) throws IOException {
		for (Subset subset : subsetDAO.getAll()) {
			log.d(TAG, "sync: sending knowledge: " + subset.getKnowledge().toString());
			communication.write(subset.getKnowledge());

			for (Metadata object : metadataDAO.getAll(subset))
				if (!containsVersion(remoteSet, object)) {
					log.d(TAG, "sync: sending metadata&data <" + object.getName() + ">");
					communication.write(object);
					communication.write(application.getObject(object.getName()));
				}

			Metadata m = new Metadata();
			log.d(TAG, "sync: sending metadata <" + m.getName() + ">");
			communication.write(m);
		}
		Subset s = new Subset();
		log.d(TAG, "sync: sending knowledge: " + s.getKnowledge().toString());
		communication.write(s.getKnowledge());
	}

	private void receiveUpdates() throws IOException, ClassNotFoundException {
		Subset remoteSet = new Subset((VersionVector) communication.read());
		log.d(TAG, "sync: received subset knowledge: " + remoteSet.getKnowledge().toString());

		while (!remoteSet.getKnowledge().isEmpty()) {
			// For every subset received, remember a set of subset-pairs resulting out of splitting sets
			Vector<Subset[]> updatedSets = new Vector<Subset[]>();
			Metadata remoteObject = (Metadata) communication.read();
			log.d(TAG, "sync: received metadata <" + remoteObject.getName() + ">");

			/*
			 * For every object received add it to (and merge the knowledge vector of the associated set into) an local
			 * set
			 */
			while (!remoteObject.getName().equals("")) {
				Object remoteData = communication.read();
				Subset localSet = containsObject(updatedSets, remoteObject.getName());

				/*
				 * If one object of the current remoteSet already forced a split, further objects associated to the
				 * original set can be added to the subset that resulted from the split
				 */
				if (localSet != null)
					merge(localSet, remoteSet, remoteObject, remoteData);
				else {
					localSet = containsObject(subsetDAO, metadataDAO, remoteObject.getName());

					if (localSet != null) {
						Subset[] splitSets = split(localSet, remoteObject);
						merge(splitSets[1], remoteSet, remoteObject, remoteData);
						updatedSets.add(splitSets);
					} else
						subsetFromNewObject(remoteSet, remoteObject, remoteData);
				}
				remoteObject = (Metadata) communication.read();
				log.d(TAG, "sync: received metadata <" + remoteObject.getName() + ">");
			}
			remoteSet = new Subset((VersionVector) communication.read());
			log.d(TAG, "sync: received subset knowledge: " + remoteSet.getKnowledge().toString());
		}
	}

	private void union() {
		boolean unifiedSth = true;

		/*
		 * Loop through all subsets and union sets with the same knowledge. Every time two sets are unified, start over
		 * again, to avoid ConcurrentModificationExceptions
		 */
		while (unifiedSth) {
			unifiedSth = false;
			Subset obsoleteSubset = null;

			for (Subset set1 : subsetDAO.getAll()) {
				set1.setObjectNames(set1.findObjectNames(metadataDAO));

				for (Subset set2 : subsetDAO.getAll()) {
					set2.setObjectNames(set2.getObjectNames());

					if (set1.getKnowledge().equals(set2.getKnowledge()) && set1.getId() != set2.getId()) {
						log.d(TAG, "sync: merging " + set1.toString() + " with " + set2.toString());

						for (Metadata object : metadataDAO.getAll(set2)) {
							set1.addObjectName(object.getName());
							metadataDAO.delete(object);
							metadataDAO.save(set1, object);
						}
						log.d(TAG, "sync: result: " + set1.toString());
						obsoleteSubset = set2;
						unifiedSth = true;
						break;
					}
					if (unifiedSth)
						break;
				}
			}
			if (unifiedSth)
				subsetDAO.delete(obsoleteSubset);
		}
	}

	private void merge(Subset localSet, Subset remoteSet, Metadata remoteObject, Object remoteData) {
		log.d(TAG, "sync: merging object " + remoteObject.toString() + " into subset "
				+ localSet.toString());
		Metadata localObject = metadataDAO.get(remoteObject.getName());

		// Choose the object with the latest version, solving conflicts if necessary
		Metadata mostRecentObject = localObject;

		if (!localSet.containsVersion(remoteObject.getVersion()))
			if (!remoteSet.containsVersion(localObject.getVersion())) {
				log.d(TAG, "sync: solving conflict on " + localObject.getName());
				mostRecentObject = application.onSynchronizationConflict(localObject, remoteObject);
			} else
				mostRecentObject = remoteObject;

		/*
		 * Merge the knowledge vector of the associated set with the one associated to the local object. If necessary,
		 * introduce versions of unknown replicas to the local vector
		 */
		Vector<Version> localKnowledge = localSet.getKnowledge();
		Vector<Version> remoteKnowledge = remoteSet.getKnowledge();

		for (int i = 0; i < remoteKnowledge.size(); ++i) {
			Version version = null;

			for (Version v : localKnowledge)
				if (v.getReplica().equals(remoteKnowledge.get(i).getReplica())) {
					version = v;
					break;
				}

			if (version == null)
				localKnowledge.add(remoteKnowledge.get(i));
			else if (remoteKnowledge.get(i).getVersion() > version.getVersion())
				version.setVersion(remoteKnowledge.get(i).getVersion());

		}
		subsetDAO.save(localSet);

		// If the remote object is the newer one save the metadata and the associated object to storage
		if (mostRecentObject == remoteObject) {
			localSet.addObjectName(remoteObject.getName());
			localObject.setVersion(remoteObject.getVersion());
			localObject.setTimestamp(remoteObject.getTimestamp());
			metadataDAO.save(localSet, localObject);
			application.putObject(localObject.getName(), remoteData);
		}
		log.d(TAG, "sync: resulting subset: " + localSet.toString());
	}

	private Subset[] split(Subset localSet, Metadata remoteObject) {
		log.d(TAG, "sync: splitting set " + localSet.toString() + " on " + remoteObject.getName());
		Subset newSet = subsetDAO.create();
		newSet.setKnowledge(localSet.getKnowledge());
		subsetDAO.save(newSet);
		Metadata object = metadataDAO.get(remoteObject.getName());
		localSet.removeObjectName(remoteObject.getName());
		metadataDAO.delete(object);
		newSet.addObjectName(remoteObject.getName());
		metadataDAO.save(newSet, object);

		if (localSet.getObjectNames().isEmpty())
			subsetDAO.delete(localSet);

		log.d(TAG, "sync: result: " + localSet.toString() + ", " + newSet.toString());

		return new Subset[] {localSet, newSet};
	}

	private void subsetFromNewObject(Subset remoteSet, Metadata remoteObject, Object remoteData) {
		Subset newSet = subsetDAO.create();
		newSet.setKnowledge(remoteSet.getKnowledge());
		newSet.addObjectName(remoteObject.getName());
		subsetDAO.save(newSet);
		metadataDAO.save(newSet, remoteObject);
		application.putObject(remoteObject.getName(), remoteData);
		log.d(TAG, "sync: created new set: " + newSet.toString());
	}
}